
package com.lee.job; 
 
import com.google.common.collect.Lists; 
import com.lee.enumeration.FinishQualifierEnum; 
import com.lee.info.*; 
import com.lee.logger.Logger; 
import com.lee.utils.*; 
import org.apache.flink.api.common.functions.FlatMapFunction; 
import org.apache.flink.util.Collector; 
 
import java.io.IOException; 
import java.time.Duration; 
import java.time.LocalDateTime; 
import java.util.ArrayList; 
import java.util.List; 
 
public class GenerateRulesFlowFunction implements FlatMapFunction<LkFlowInfo, List<RuleFlowInfo>> { 
 
    long index;
    long cIndex;
 
 
    @Override
    public void flatMap(LkFlowInfo lkFlowInfo, Collector<List<RuleFlowInfo>> collector)  {
        try {
            if(DMapUtils.getIndex() > ConstantsUtils.DATA_BLOCK_SIZE){
                Logger.debug(lkFlowInfo.getDIndex(),"/",lkFlowInfo.getDCIndex(),"GenerateRulesFlowFunction -> 正在计算汇总结果，退出当前模块  ");
                return;
            }
            Logger.debug(lkFlowInfo.getDIndex(),"/",lkFlowInfo.getDCIndex(),"GenerateRulesFlowFunction -> lkFlowInfo ", lkFlowInfo);
            LocalDateTime start = LocalDateTime.now();
            List<RuleFlowInfo> bigRuleList = new ArrayList<>();
            List<String> LFlow = lkFlowInfo.getLkList();
            index = lkFlowInfo.getDIndex();
            cIndex = lkFlowInfo.getDCIndex();
 
            List<String> freqSet = LFlow;
            if (freqSet.size() < 2) {
                return;
            }
            List<List<String>> H1 = new ArrayList<>();
            for (String item : freqSet) {
                H1.add(Lists.newArrayList(item));
            }
            LkFlowInfos H1LkFlowInfo = new LkFlowInfos();
            H1LkFlowInfo.setLkList(H1);
 
            calcConf(freqSet, H1LkFlowInfo, bigRuleList);
            rulesFromConseq(freqSet, H1LkFlowInfo, bigRuleList);
 
 
            Logger.debug(index, "BdsGenerateRulesFlowFunction -> big rule list ", bigRuleList);
            collector.collect(bigRuleList);
 
            DMapUtils.addFinishIndex(FinishQualifierEnum.GENERATE_RULES.getCode(),1l);
            LocalDateTime end = LocalDateTime.now();
            Duration between = Duration.between(start, end);
            Logger.debug(lkFlowInfo.getDIndex(),"/",lkFlowInfo.getDCIndex(),between.toMillis(),"END GenerateRulesFlowFunction -> lkFlowInfo ", lkFlowInfo);
 
        }catch(Exception e){
            Logger.error("BdsGenerateRulesFlowFunction -> 产生规则报错：",e);
            e.printStackTrace();
        }
    }
 
    private void rulesFromConseq(List<String> freqSet, LkFlowInfos lkFlowInfo, List<RuleFlowInfo> brl) throws IOException {
        List<List<String>> H  = lkFlowInfo.getLkList();
        int m = H.get(0).size();
        if(freqSet.size() > (m+1)){
            CkFlowInfos ckFlowInfo = aprioriGen(lkFlowInfo, m+1);
            List<List<String>> Hmp1 = ckFlowInfo.getCkList();
            LkFlowInfos HmpLkFlowInfo = new LkFlowInfos();
            HmpLkFlowInfo.setLkList(Hmp1);
            Hmp1 = calcConf(freqSet, HmpLkFlowInfo,brl);
            if(Hmp1.size() > 0){
                HmpLkFlowInfo.setLkList(Hmp1);
                rulesFromConseq(freqSet,HmpLkFlowInfo,brl);
            }
 
        }
    }
    private List<List<String>> calcConf(List<String> freqSet, LkFlowInfos lkFlowInfo, List<RuleFlowInfo> brl) throws IOException {
 
        List<List<String>> H = lkFlowInfo.getLkList();
        List<List<String>> prunedH = new ArrayList<>();
        double freqConf = SupportDataUtils.getValueByKey(freqSet);
        long freqNum = SupportDataUtils.getValueByKeyNum(freqSet);
        Logger.debug(index,"6 BdsGenerateRulesFlowFunction rulesFromConseq -> H ");
        for(List<String> conseq:H){
            List<String> subItem = ArrayUtils.getSubItem(freqSet,conseq);
            Logger.debug(index,cIndex,"7 BdsGenerateRulesFlowFunction calcConf -> freqSet, conseq, subItem",freqSet, conseq, subItem);
            long subNum = SupportDataUtils.getValueByKeyNum(subItem);
            if(subNum == 0){
                //除数为0处理
                continue;
            }
            double conf = 1.0 * freqNum / subNum;
            Logger.debug(index,cIndex,"8 BdsGenerateRulesFlowFunction calcConf -> check (",subItem," --> ",conseq," conf ",conf,",num",freqNum,")");
            if(conf >= ConstantsUtils.MIN_CONF){
 
                Logger.info(index,cIndex,"9 BdsGenerateRulesFlowFunction calcConf -> got (",subItem," --> ",conseq," conf ",conf,",num",freqNum,")");
                RulesUtils.setValueByKey(cIndex,subItem,conseq,conf,freqConf);
 
                RuleFlowInfo ruleFlowInfo = new RuleFlowInfo();
                ruleFlowInfo.setFrom(subItem);
                ruleFlowInfo.setTo(conseq);
                ruleFlowInfo.setWeight(conf);
                brl.add(ruleFlowInfo);
                prunedH.add(conseq);
            }
        }
        return prunedH;
 
    }
    private CkFlowInfos aprioriGen(LkFlowInfos lkFlowInfo, int k){
        List<List<String>> Lk_1 = lkFlowInfo.getLkList();
        List<List<String>> Ck = new ArrayList<>();
        int lenLk = Lk_1.size();
        Logger.debug(index,"4 BdsGenerateRulesFlowFunction rulesFromConseq -> H ");
        for(int i=0;i<lenLk;i++){
            List<String> L1 = Lk_1.get(i).subList(0,k - 2);
            L1.sort(String::compareTo);
            for(int j=i+1;j<lenLk;j++){
                List<String> L2 = Lk_1.get(j).subList(0,k - 2);
                L2.sort(String::compareTo);
                if(ArrayUtils.isEqual(L1,L2)){
                    Ck.add(ArrayUtils.getUniqItem(Lk_1.get(i),Lk_1.get(j)));
                }
            }
        }
        Logger.debug(index,"5 BdsGenerateRulesFlowFunction rulesFromConseq -> H ");
        CkFlowInfos ckFlowInfo = new CkFlowInfos();
        ckFlowInfo.setCkList(Ck);
        ckFlowInfo.setDindex(index);
        return ckFlowInfo;
 
    }
 
} 
